package com.zzl.rabbitmq.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.zzl.rabbitmq.utils.RabbitMqUtils;

import java.nio.charset.StandardCharsets;

/**
 * Title: ComfirmMessage.class<br>
 * Description: <br>
 * Copyright (c)  2021    <br>
 * Create DateTime: 2021/7/7 16:31 <br>
 *
 * @author zhuangzhuangliang
 */
public class ConfirmMessage {

    public static final int MESSAGE_COUNT = 1000;
    public static final String QUEUE_NAME = "confirm_queue";

    public static void main(String[] args) throws Exception {
//        publishMessageIndividual(); //耗时 : 耗时 : 1102
//        publishMessageBatch();//耗时 : 233

        publishMessageASync(); //耗时 : 126
    }

    public static Channel init() {
        Channel channel = null;
        try {
            channel = RabbitMqUtils.getChannel();
            //使用发布确认机制
            channel.confirmSelect();
            //声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return channel;
    }

    /**
     * 单个消息确认
     *
     * @throws Exception
     */
    public static void publishMessageIndividual() throws Exception {

        Channel channel = init();
        long begin = System.currentTimeMillis();
        //发布1000消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String msg = i + "";
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes(StandardCharsets.UTF_8));
            channel.waitForConfirms();
        }

        System.out.println("耗时 : " + (System.currentTimeMillis() - begin));
    }

    /**
     * 批量确认消息
     *
     * @throws Exception
     */
    public static void publishMessageBatch() throws Exception {
        Channel channel = init();
        long begin = System.currentTimeMillis();
        //发布1000消息
        int batchSize = 100;
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String msg = i + "";
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes(StandardCharsets.UTF_8));
            if (i % batchSize == 0) {
                channel.waitForConfirms();
            }
        }
        System.out.println("耗时 : " + (System.currentTimeMillis() - begin));
    }

    /**
     * 异步确认消息
     */
    public static void publishMessageASync() throws Exception {
        Channel channel = init();
        long begin = System.currentTimeMillis();


        ConfirmCallback ackConfirmCallback = (deliveryTag, multiple) -> {
            System.out.println("消息确认成功: "+deliveryTag);
        };

        ConfirmCallback nackConfirmCallback =  (deliveryTag, multiple) -> {
            System.out.println("消息确认失败: "+deliveryTag);
        };

        //注册消息确认监听器
        channel.addConfirmListener(ackConfirmCallback,nackConfirmCallback);
        //发布1000消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String msg = i + "";
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes(StandardCharsets.UTF_8));
        }
        System.out.println("耗时 : " + (System.currentTimeMillis() - begin));
    }
}
